package com.mall.promo.mq;

import com.alibaba.fastjson.JSON;
import com.mall.order.dto.CreateSeckillOrderRequest;
import com.mall.promo.cache.CacheManager;
import com.mall.promo.dal.entitys.PromoItem;
import com.mall.promo.dal.persistence.PromoItemMapper;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.*;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.common.message.MessageExt;
import org.redisson.api.RLock;
import org.redisson.api.RedissonClient;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;
import tk.mybatis.mapper.entity.Example;

import javax.annotation.PostConstruct;
import java.util.HashMap;
import java.util.Map;
import java.util.concurrent.TimeUnit;

@Component
@Transactional
public class PromoOrderProducer {

    private TransactionMQProducer producer;
    @Autowired
    CacheManager cacheManager;
    @Autowired
    RedissonClient redissonClient;

    @Autowired
    PromoItemMapper promoItemMapper;
    @PostConstruct
    public void init(){
        producer = new TransactionMQProducer("promo-order-producer");
        producer.setNamesrvAddr("127.0.0.1:9876");

        producer.setTransactionListener(new TransactionListener() {
            @Override
            public LocalTransactionState executeLocalTransaction(Message message, Object o) {
                //执行本地事务,扣减秒杀库存
                Map<String,Long> promoMap = (Map<String, Long>) o;
                Long psId = promoMap.get("psId");
                Long productId = promoMap.get("productId");
                String transactionId = message.getTransactionId();
                //使用redis客户端对指定秒杀场次的商品库存加锁,可重入锁
                String lockKey = "promo_order_lock"+psId+productId;
                RLock lock = redissonClient.getLock(lockKey);
                //锁10秒过期
                lock.lock(10, TimeUnit.SECONDS);
                //查询剩余库存,库存为空存入库存为空的信息
                Example example = new Example(PromoItem.class);
                example.createCriteria().andEqualTo("psId",psId).andEqualTo("itemId",productId);
                PromoItem promoItem = promoItemMapper.selectOneByExample(example);
                if(promoItem.getItemStock() < 1) {
                    cacheManager.setCache("promo_stock_" + psId + "_" + productId, "true", 1);
                }
                try {
                    //秒杀库存减1
                    Integer effectedRow = promoItemMapper.decreaseStock(productId, psId);
                    //获取记录本地事务是否执行成功的key
                    String localTransactionKey = "promo_oder_transaction:"+transactionId;
                    if(effectedRow<1){
                        //扣减库存库存失败,事务失败,将本地事务状态存入redis,有效期1天
                        cacheManager.setCache(localTransactionKey,"fail",1);
                        //告知服务器本地事务回滚
                        return LocalTransactionState.ROLLBACK_MESSAGE;
                    }
                    //扣减库存成功,订阅组消费信息,创建订单
                    cacheManager.setCache(localTransactionKey,"success",1);
                    return LocalTransactionState.COMMIT_MESSAGE;
                }finally {
                    lock.unlock();
                }
            }

            @Override
            public LocalTransactionState checkLocalTransaction(MessageExt messageExt) {
                //MQ检查本地事务是否完成
                String transactionId = messageExt.getTransactionId();
                String localTransactionKey = "promo_oder_transaction:"+transactionId;
                String checkCache = cacheManager.checkCache(localTransactionKey);
                //事务状态数据为空,本地事务还未执行完成,返回状态未知,MQ会稍后再查
                if(checkCache == null) return LocalTransactionState.UNKNOW;
                //失败回滚
                if(checkCache.equals("fail")) return LocalTransactionState.ROLLBACK_MESSAGE;
                //成功消费
                return LocalTransactionState.COMMIT_MESSAGE;
            }
        });
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }
    //传入创建秒杀订单的参数,和扣减秒杀库存本地事务的参数
    public boolean sendMsg(CreateSeckillOrderRequest request,Long psId,Long productId){
        //下单的信息作为消息体
        String body = JSON.toJSONString(request);
        Message message = new Message("createPromoOrder", "orderMsg2", body.getBytes());
        //传入秒杀的信息
        Map<String, Long> promoMap = new HashMap<>();
        promoMap.put("psId",psId);
        promoMap.put("productId",productId);
        TransactionSendResult sendResult = null;
        try {
            //传入秒杀参数,记录本地事务执行完成情况
            sendResult = producer.sendMessageInTransaction(message, promoMap);
        } catch (MQClientException e) {
            e.printStackTrace();
        }
        //返回下单信息是否发给Mq服务器
        return sendResult!=null&&LocalTransactionState.COMMIT_MESSAGE.equals(sendResult.getLocalTransactionState());

    }
}
